const { Transform } = require("stream");

class BackPressureTransform extends Transform {
  constructor(option = {}) {
    option = {
      readableObjectMode: true,
      writableObjectMode: true,
      highWaterMark: 1,
      ...option,
    };

    super(option);
    this.name = option.name;
    let oldPipe = this.pipe.bind(this);
    this.pipe = function (dest, option) {
      dest.on("drain", () => {
        this.emit("destDrain");
      });
      this.dest = dest;
      dest.src = this;
      return oldPipe(dest, option);
    };
  }
  backPressurePush(arr, callback) {
    let ret = this.push(arr);
    this.onDrain = () => {
      callback();
    };
    if (ret) {
      let onDrain = this.onDrain;
      this.onDrain = undefined;
      onDrain && onDrain(); // 必须设置undefined然后运行，因为运行后就进入下一个step了，this.onDrain就被设置成新的，然后这里的undefined就会覆盖
    } else {
      this.once("destDrain", () => {
        process.nextTick(() => {
          // 如果不是下一个tick运行的话可能会卡住。。
          let onDrain = this.onDrain;
          this.onDrain = undefined;
          onDrain && onDrain(); // 必须设置undefined然后运行，因为运行后就进入下一个step了，this.onDrain就被设置成新的，然后这里的undefined就会覆盖
        });
      });
    }
  }
}

module.exports = BackPressureTransform;
